package cn.xval.ft.core.marketdata.manage;

import cn.xval.ft.core.ctp.CThostFtdcDepthMarketDataField;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class SpiManager {
    private final static ConcurrentHashMap<String, MarketDataSpi> spis = new ConcurrentHashMap<>();

    private enum WatchingState { pending, tobeWatching, watching }
    private static class WatchingItemInfo{
        String connectId;
        WatchingState state;
        Date begin;

        public WatchingItemInfo(String connectId, WatchingState state, Date begin) {
            this.connectId = connectId;
            this.state = state;
            this.begin = begin;
        }

        public boolean isExpired() {
            return state.equals(WatchingState.tobeWatching) && (new Date().getTime() - begin.getTime() >= 1000);
        }
    }
    private final static ConcurrentHashMap<String, WatchingItemInfo> watchingContractCodes = new ConcurrentHashMap<>();
    public static void addSpi(MarketDataSpi spi) {
        spis.put(spi.getId(), spi);
    }

    public static void updateWatchingState(String contractCode, boolean success) {
        var wc = watchingContractCodes.get(contractCode);
        var state = success ? WatchingState.watching : WatchingState.pending;
        watchingContractCodes.put(contractCode, new WatchingItemInfo(wc.connectId, state, new Date()));
    }

    public static void subscribeMarketData(String connectId, List<String> contractCodes) {
        if (contractCodes.isEmpty())
            return;
        List<String> tobeRemove = new ArrayList<>();
        contractCodes.forEach(cc -> {
            var a = watchingContractCodes.get(cc);
            if (a != null) {
                if (a.state == WatchingState.watching || a.state == WatchingState.tobeWatching) {
                    // 已经在监听，不需要任何操作，删除
                    tobeRemove.add(cc);
                }
                else if (!a.connectId.equals(connectId)) {
                    // 等待或发生错误，链接不同，更换链接
                    watchingContractCodes.remove(cc);
                }
            }
        });
        tobeRemove.forEach(contractCodes::remove);

        var spi = spis.get(connectId);
        if (spi == null || !spi.getStatus().getStatus().isActive()) {
            // 链接不可用，阻塞等待
            contractCodes.forEach(cc -> {
                watchingContractCodes.put(cc, new WatchingItemInfo(connectId, WatchingState.pending, new Date()));
            });
            return;
        }
        contractCodes.forEach(cc -> {
            watchingContractCodes.put(cc, new WatchingItemInfo(connectId, WatchingState.tobeWatching, new Date()));
        });
        spi.subscribe(contractCodes);
    }

    public static void connectActive(String connectId) {
        List<String> pendingContractCodes = new ArrayList<>();

        watchingContractCodes.forEachEntry(500, en -> {
            if (en.getValue().connectId.equals(connectId)
                    && (en.getValue().state == WatchingState.pending || en.getValue().isExpired())) {
                pendingContractCodes.add(en.getKey());
            }
        });
        pendingContractCodes.forEach(watchingContractCodes::remove);

        subscribeMarketData(connectId, pendingContractCodes);
    }

    private final static ConcurrentLinkedDeque<DataWatcher> watchers = new ConcurrentLinkedDeque<>();
    public static void removeDataWatcher(DataWatcher watcher) {
        watchers.remove(watcher);
    }

    /**
     * 添加监视
     * @param queue
     */
    public static void addDataWatcher(DataWatcher watcher) {
        watchers.add(watcher);
    }

    public static void dispatchMarketData(CThostFtdcDepthMarketDataField data) {
        watchers.forEach(w -> {
            if (w.post(data) >= 10) {
                // 多于 10 错误
                w.remove();
            }
        });
    }

    private static final AtomicInteger id = new AtomicInteger(0);
    public static int nextRequestId() {
        return id.addAndGet(1);
    }
}
